package log.job;

import log.function.StockTickSourceFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.Timestamp;

/**
 * 股票涨跌实时监控CEP实现
 * 规则：5分钟内涨跌幅超过3%且短期波动率大于2%
 */
public class StockRiseFallCEPJob {

    private static final int DECIMAL_SCALE = 4;
    private static final RoundingMode ROUNDING_MODE = RoundingMode.HALF_UP;

    public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);

        // 1. 从数据源读取股票实时数据
        DataStream<StockTick> stockStream = env
                .addSource(new StockTickSourceFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<StockTick>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((event, timestamp) -> event.getEventTime())
                );

        // 2. 定义CEP模式
        Pattern<StockTick, StockTick> pattern = Pattern.<StockTick>begin("start")
                .next("alertCheck")
                .where(new SimpleCondition<StockTick>() {
                    @Override
                    public boolean filter(StockTick currentTick) {
                        try {
                            // 计算涨跌幅: (当前成交价 - 前一个窗口收盘价) / 前一个窗口收盘价 * 100%
                            BigDecimal priceChangeRate = calculatePriceChangeRate(
                                    currentTick.getCurrentPrice(),
                                    currentTick.getPrevWindowClosePrice()
                            );

                            // 计算短期波动率: (当前价格 - 上一个价格) / 上一个价格 * 100%
                            BigDecimal shortTermVolatility = calculateShortTermVolatility(
                                    currentTick.getCurrentPrice(),
                                    currentTick.getLastPrice()
                            );

                            // 判断是否同时满足两个条件
                            boolean isPriceChangeAlert = priceChangeRate.compareTo(new BigDecimal("3")) > 0;
                            boolean isVolatilityAlert = shortTermVolatility.compareTo(new BigDecimal("2")) > 0;

                            return isPriceChangeAlert && isVolatilityAlert;
                        } catch (Exception e) {
                            e.printStackTrace(); // 实际项目中应使用日志框架
                            return false;
                        }
                    }
                })
                .within(Time.minutes(5));

        // 3. 应用CEP模式
        PatternStream<StockTick> patternStream = CEP.pattern(
                stockStream.keyBy(StockTick::getStockCode),
                pattern
        );

        // 4. 处理匹配的预警事件
        DataStream<AlertResult> alertStream = patternStream.process(
                new PatternProcessFunction<StockTick, AlertResult>() {
                    @Override
                    public void processMatch(Map<String, List<StockTick>> map, Context context, Collector<AlertResult> collector) throws Exception {
                        List<StockTick> alertTicks = map.get("alertCheck");
                        if (alertTicks != null && !alertTicks.isEmpty()) {
                            StockTick tick = alertTicks.get(0);
                            try {
                                BigDecimal priceChangeRate = calculatePriceChangeRate(tick.getCurrentPrice(), tick.getPrevWindowClosePrice());
                                BigDecimal shortTermVolatility = calculateShortTermVolatility(tick.getCurrentPrice(), tick.getLastPrice());

                                AlertResult alert = new AlertResult(
                                        tick.getTradeDate(),
                                        tick.getStockCode(),
                                        new Timestamp(System.currentTimeMillis()),
                                        tick.getCurrentPrice(),
                                        tick.getPrevWindowClosePrice(),
                                        priceChangeRate,
                                        shortTermVolatility,
                                        true
                                );
                                collector.collect(alert);
                            } catch (Exception e) {
                                e.printStackTrace(); // 实际项目中应使用日志框架
                            }
                        }
                    }


                }
        );

        // 5. 将预警结果写入Doris
        alertStream.addSink(new DorisSinkFunction());

        // 6. 执行任务
        env.execute("Stock Rise Fall Real-time Monitoring");
    }

    // 计算涨跌幅
    private static BigDecimal calculatePriceChangeRate(BigDecimal currentPrice, BigDecimal prevClosePrice) {
        if (prevClosePrice == null || prevClosePrice.compareTo(BigDecimal.ZERO) == 0) {
            return BigDecimal.ZERO;
        }
        return currentPrice.subtract(prevClosePrice)
                .divide(prevClosePrice, DECIMAL_SCALE, ROUNDING_MODE)
                .multiply(new BigDecimal("100"));
    }

    // 计算短期波动率
    private static BigDecimal calculateShortTermVolatility(BigDecimal currentPrice, BigDecimal lastPrice) {
        if (lastPrice == null || lastPrice.compareTo(BigDecimal.ZERO) == 0) {
            return BigDecimal.ZERO;
        }
        return currentPrice.subtract(lastPrice)
                .divide(lastPrice, DECIMAL_SCALE, ROUNDING_MODE)
                .multiply(new BigDecimal("100"));
    }

    // 股票行情数据类
    public static class StockTick {
        private String tradeDate;
        private String stockCode;
        private BigDecimal currentPrice;
        private BigDecimal prevWindowClosePrice;
        private BigDecimal lastPrice;
        private Long eventTime;

        // getters and setters
        public String getTradeDate() { return tradeDate; }
        public void setTradeDate(String tradeDate) { this.tradeDate = tradeDate; }

        public String getStockCode() { return stockCode; }
        public void setStockCode(String stockCode) { this.stockCode = stockCode; }

        public BigDecimal getCurrentPrice() { return currentPrice; }
        public void setCurrentPrice(BigDecimal currentPrice) { this.currentPrice = currentPrice; }

        public BigDecimal getPrevWindowClosePrice() { return prevWindowClosePrice; }
        public void setPrevWindowClosePrice(BigDecimal prevWindowClosePrice) { this.prevWindowClosePrice = prevWindowClosePrice; }

        public BigDecimal getLastPrice() { return lastPrice; }
        public void setLastPrice(BigDecimal lastPrice) { this.lastPrice = lastPrice; }

        public Long getEventTime() { return eventTime; }
        public void setEventTime(Long eventTime) { this.eventTime = eventTime; }
    }

    // 预警结果类
    public static class AlertResult {
        private String tradeDate;
        private String stockCode;
        private Timestamp windowEndTime;
        private BigDecimal currentPrice;
        private BigDecimal prevWindowClosePrice;
        private BigDecimal priceChangeRate;
        private BigDecimal shortTermVolatility;
        private Boolean isAlertCondition;
        private Timestamp updateTime;

        public AlertResult(String tradeDate, String stockCode, Timestamp windowEndTime,
                           BigDecimal currentPrice, BigDecimal prevWindowClosePrice,
                           BigDecimal priceChangeRate, BigDecimal shortTermVolatility,
                           Boolean isAlertCondition) {
            this.tradeDate = tradeDate;
            this.stockCode = stockCode;
            this.windowEndTime = windowEndTime;
            this.currentPrice = currentPrice;
            this.prevWindowClosePrice = prevWindowClosePrice;
            this.priceChangeRate = priceChangeRate;
            this.shortTermVolatility = shortTermVolatility;
            this.isAlertCondition = isAlertCondition;
            this.updateTime = new Timestamp(System.currentTimeMillis());
        }

        // getters
        public String getTradeDate() { return tradeDate; }
        public String getStockCode() { return stockCode; }
        public Timestamp getWindowEndTime() { return windowEndTime; }
        public BigDecimal getCurrentPrice() { return currentPrice; }
        public BigDecimal getPrevWindowClosePrice() { return prevWindowClosePrice; }
        public BigDecimal getPriceChangeRate() { return priceChangeRate; }
        public BigDecimal getShortTermVolatility() { return shortTermVolatility; }
        public Boolean getIsAlertCondition() { return isAlertCondition; }
        public Timestamp getUpdateTime() { return updateTime; }
    }

    // Doris Sink Function
    public static class DorisSinkFunction extends org.apache.flink.streaming.api.functions.sink.RichSinkFunction<AlertResult> {
        private static final String DORIS_URL = System.getenv("DORIS_URL") != null ? System.getenv("DORIS_URL") : "jdbc:mysql://node102:9030/gd1_gp";
        private static final String USERNAME = System.getenv("DORIS_USERNAME") != null ? System.getenv("DORIS_USERNAME") : "root";
        private static final String PASSWORD = System.getenv("DORIS_PASSWORD") != null ? System.getenv("DORIS_PASSWORD") : "123456";

        private static final String INSERT_SQL = "INSERT INTO realtime_rise_fall_stock " +
                "(trade_date, stock_code, window_end_time, current_price, prev_window_close_price, " +
                "price_change_rate, short_term_volatility, is_alert_condition, update_time) " +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";

        @Override
        public void invoke(AlertResult value, Context context) throws Exception {
            try (Connection conn = DriverManager.getConnection(DORIS_URL, USERNAME, PASSWORD);
                 PreparedStatement pstmt = conn.prepareStatement(INSERT_SQL)) {

                pstmt.setString(1, value.getTradeDate());
                pstmt.setString(2, value.getStockCode());
                pstmt.setTimestamp(3, value.getWindowEndTime());
                pstmt.setBigDecimal(4, value.getCurrentPrice());
                pstmt.setBigDecimal(5, value.getPrevWindowClosePrice());
                pstmt.setBigDecimal(6, value.getPriceChangeRate());
                pstmt.setBigDecimal(7, value.getShortTermVolatility());
                pstmt.setBoolean(8, value.getIsAlertCondition());
                pstmt.setTimestamp(9, value.getUpdateTime());

                pstmt.executeUpdate();
            }
        }
    }
}
